
#include "stdafx.h"
#include "TcpPortTestActiveConnectionActor.h"
#include "TcpPortTestApp.h"

#define ERROR_CLIENTCONNECTIONACTOR_ONREAD_UNEXPECTED_ERROR		-10001
#define ERROR_CLIENTCONNECTIONACTOR_ONWRITE_UNEXPECTED_ERROR	-10002
#define ERROR_CLIENTCONNECTIONACTOR_ONCONNECT_UNEXPECTED_ERROR	-10003

TcpPortTestActiveConnectionActorMgr::TcpPortTestActiveConnectionActorMgr( std::size_t _workerPoolSize /*= 1*/ )
:ActorMgrBase(_workerPoolSize)
{

}

TcpPortTestActiveConnectionActorMgr::~TcpPortTestActiveConnectionActorMgr()
{

}

retT TcpPortTestActiveConnectionActorMgr::makeActor( boost::shared_ptr<ActorBase>& _pActor )
{
	_pActor.reset(new TcpPortTestActiveConnectionActor());
	R(ActorMgrBase::makeActor(_pActor));
	RETSUCCESS;
}

retT TcpPortTestActiveConnectionActorMgr::start()
{
	boost::shared_ptr<ActorBase> pActor;
	R(makeActor(pActor));
	boost::dynamic_pointer_cast<TcpPortTestActiveConnectionActor>(pActor)->numToEcho_ = ((TcpPortTestConfigure*)getApp()->getConfigure())->numToEcho_;
	R(ActorMgrBase::start());
	RETSUCCESS;
}

TcpPortTestActiveConnectionActor::TcpPortTestActiveConnectionActor()
:serverIP_(),serverPort_(0),doConnect_(false),numToEcho_(0),pTcpConnectionHandler()
{

}

TcpPortTestActiveConnectionActor::~TcpPortTestActiveConnectionActor()
{
	numToEcho_ = 0;
	serverPort_ = 0;
	doConnect_ = false;
	pTcpConnectionHandler.reset();
}

retT TcpPortTestActiveConnectionActor::start()
{
	R(ActorBase::start());
	if(doConnect_)
	{
		boost::shared_ptr<DataHandlerContext> pDataHandlerContext(new DataHandlerContext);
		pDataHandlerContext->pTimerDuration_.reset(new boost::posix_time::seconds(3));
		R(pTcpConnectionHandler->asyncConnect(
			serverIP_,
			serverPort_,
			pDataHandlerContext));
	}
	RETSUCCESS;
}

retT TcpPortTestActiveConnectionActor::init()
{
	R(ActorBase::init());
	pTcpConnectionHandler.reset(new TcpHandler());
	boost::shared_ptr<boost::asio::ip::tcp::endpoint> pLocalEndpoint(new boost::asio::ip::tcp::endpoint(
		boost::asio::ip::address::from_string(((TcpPortTestConfigure*)getApp()->getConfigure())->listeningIP_),
		((TcpPortTestConfigure*)getApp()->getConfigure())->listeningPort_));
	pTcpConnectionHandler->setLocalEndpoint(pLocalEndpoint);
	pTcpConnectionHandler->attachHandlerOwner(shared_from_this());
	serverIP_ = ((TcpPortTestConfigure*)getApp()->getConfigure())->serverIP_;
	serverPort_ = ((TcpPortTestConfigure*)getApp()->getConfigure())->serverPort_;
	doConnect_ = ((TcpPortTestConfigure*)getApp()->getConfigure())->doConnect_;
	RETSUCCESS;
}

retT TcpPortTestActiveConnectionActor::onRead( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
{
	R(ITcpAction::onRead(_pDataHandlerContext, _bytesTransferred, e, _timeoutCount));
	if(!e)
	{
		if(_pDataHandlerContext)
		{
			int i;
			R(_pDataHandlerContext->pBuffer_->read(i));
			if(i != numToEcho_)
			{
				LOG_ERROR(boost::format( "Wrong echo number %1% in %2% %3% %4%" ) % i % __FUNCTION__ % __FILE__ % __LINE__);
				R(close());
			}
			std::cout << i << std::endl;
			R(_pDataHandlerContext->pBuffer_->seek(0));
			_pDataHandlerContext->bytesToTransfer_ = 0;
			R(pTcpConnectionHandler->asyncWrite(_pDataHandlerContext->pBuffer_, _pDataHandlerContext));
		}
	}
	else if(e == boost::asio::error::eof || e == boost::asio::error::connection_aborted
		|| e == boost::asio::error::connection_reset || e == boost::asio::error::connection_refused
		|| e == boost::asio::error::shut_down)
	{
		LOG_INFO(boost::format( "errorNo %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		R(close());
		if(boost::shared_ptr<ActorMgrBase> pActorMgr = pActorMgr_.lock())
		{
			R(pActorMgr->post(boost::bind(&ActorMgrBase::checkToStopApp, pActorMgr)));
		}
	}
	else
	{
		LOG_ERROR(boost::format( "unexpected error %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(ERROR_CLIENTCONNECTIONACTOR_ONREAD_UNEXPECTED_ERROR);
	}
	RETSUCCESS;
}

retT TcpPortTestActiveConnectionActor::onWrite( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
{
	R(ITcpAction::onWrite(_pDataHandlerContext, _bytesTransferred, e, _timeoutCount));
	if(!e)
	{
		if(_pDataHandlerContext)
		{
			_pDataHandlerContext->pBuffer_->clear();
			_pDataHandlerContext->bytesToTransfer_ = sizeof(int);
			R(pTcpConnectionHandler->asyncRead(_pDataHandlerContext->pBuffer_, _pDataHandlerContext));
		}
	}
	else if(e == boost::asio::error::eof || e == boost::asio::error::connection_aborted
		|| e == boost::asio::error::connection_reset || e == boost::asio::error::connection_refused
		|| e == boost::asio::error::shut_down)
	{
		LOG_INFO(boost::format( "errorNo %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		R(close());
		if(boost::shared_ptr<ActorMgrBase> pActorMgr = pActorMgr_.lock())
		{
			R(pActorMgr->post(boost::bind(&ActorMgrBase::checkToStopApp, pActorMgr)));
		}
	}
	else
	{
		LOG_ERROR(boost::format( "unexpected error %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(ERROR_CLIENTCONNECTIONACTOR_ONWRITE_UNEXPECTED_ERROR);
	}
	RETSUCCESS;
}

retT TcpPortTestActiveConnectionActor::onConnect( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, const boost::system::error_code& e, std::size_t _timeoutCount )
{
	R(ITcpAction::onConnect(_pDataHandlerContext, e, _timeoutCount));
	if(!e)
	{
		if(_timeoutCount == 0)
		{
			if(_pDataHandlerContext)
			{
				R(_pDataHandlerContext->reset());
				if(boost::shared_ptr<WorkerBase> pWorker = getWorker())
				{
					boost::shared_ptr<IoBuffer> pBuffer(new IoBuffer(pWorker));
					R(pBuffer->write(numToEcho_));
					R(pTcpConnectionHandler->asyncWrite(pBuffer, _pDataHandlerContext));
				}
				else
					RET(ERROR_COMMON_SHARED_PTR_NULL);
			}
		}
		else
		{
			R(pTcpConnectionHandler->asyncConnect(
				serverIP_,
				serverPort_,
				_pDataHandlerContext));
		}
	}
	else if(e == boost::asio::error::eof || e == boost::asio::error::connection_aborted
		|| e == boost::asio::error::connection_reset || e == boost::asio::error::connection_refused
		|| e == boost::asio::error::shut_down)
	{
		//R(pTcpConnectionHandler->reset());
		//R(pTcpConnectionHandler->asyncConnect(
		//	serverIP_,
		//	serverPort_,
		//	_pDataHandlerContext));
	}
	else
	{

		LOG_ERROR(boost::format( "unexpected error %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(ERROR_CLIENTCONNECTIONACTOR_ONCONNECT_UNEXPECTED_ERROR);
	}
	RETSUCCESS;
}

retT TcpPortTestActiveConnectionActor::onError( retT errorNo )
{
	errorNo = ActorBase::onError(errorNo);
	if(boost::shared_ptr<ActorMgrBase> pActorMgr = pActorMgr_.lock())
	{
		pActorMgr->post(boost::bind(&ActorMgrBase::checkToStopApp, pActorMgr));
	}
	return errorNo;
}
